package main

import (
	"context"
	"encoding/json"
	"fmt"
	"gopkg.in/yaml.v2"
	"lifi"
	"lifi/config"
	"lifi/internal"
	"lifi/models"
	"lifi/plugins/sinks"
	"lifi/plugins/sources"
	"lifi/plugins/transfers"
	"log"
	"runtime"
	"sort"
	"sync"
	"time"
)

type Pipe struct {
	Pipeline config.Pipeline `json:"pipelines" yaml:"pipelines"`
}
type sourceUnit struct {
	dst    chan<- lifi.Data
	inputs []models.RunningSource
}
type transferUnit struct {
	src      <-chan lifi.Data
	transfer *models.RunningTransfer
	dst      chan<- lifi.Data
}

type sinkUnit struct {
	src   <-chan lifi.Data
	sinks []models.RunningSink
	trans map[string]transferUnits
}
type transferUnits []*transferUnit

func (a *Pipe) RunPipeline(ctx context.Context) error {

	// run inputs
	next, ou, err := a.startSink(ctx, a.Pipeline.Sinks)
	if err != nil {
		log.Printf("strat sink error:%v\n", err)
		return err
	}

	/*
		var tu []*transferUnit
		if len(a.Pipeline.Transfers) != 0 {
			next, tu, err = a.startTransfer(next, a.Pipeline.Transfers)
			if err != nil {
				return err
			}
		}

	*/

	iu, err := a.startSource(next, a.Pipeline.Source)
	if err != nil {
		log.Printf("start source error:%v\n", err)
		return err
	}

	// run

	var wg sync.WaitGroup

	wg.Add(1)
	go func() {
		defer wg.Done()
		a.runSinks(ou)
	}()

	startTime := time.Now()
	wg.Add(1)
	go func() {
		defer wg.Done()
		a.runSources(ctx, startTime, iu)
	}()

	fmt.Printf("iu.dst:%p\n", iu.dst)
	for _, v := range ou.trans {
		for _, vv := range v {
			fmt.Printf("ou.trans[%v]->src:%p\n", vv.transfer.Name, vv.src)
			fmt.Printf("ou.trans[%v]->dst:%p\n", vv.transfer.Name, vv.dst)
		}
	}
	fmt.Printf("ou.src:%p\n", ou.src)

	wg.Wait()
	log.Printf("Pipeline finished:%v\n", err)
	return err
}
func (a *Pipe) connectSink(ctx context.Context, sink models.RunningSink) error {

	err := sink.Sink.Connect()
	if err != nil {
		log.Printf("Failed to connect to sink [%s], retrying in 15s, "+
			"error was %q\n", sink.Name, err)
		err = internal.SleepContext(ctx, 15*time.Second)
		if err != nil {
			log.Printf("sleepContex error:%v\n", err)
			return err
		}
		err = sink.Sink.Connect()
		if err != nil {
			return fmt.Errorf("failed to connect to [%s], error was %q", sink.Name, err)
		}
	}
	return nil

}

func (a *Pipe) startSource(
	dst chan<- lifi.Data,
	input models.RunningSource,
) (*sourceUnit, error) {

	unit := &sourceUnit{
		dst: dst,
	}
	creator, ok := sources.Sources[input.Name]
	if !ok {
		log.Printf("undefined but requested source: %s", input.Name)
	}
	source := creator()
	a.UnmarshalConfig(*input.Config, source)

	//若是StreamingSource 有 Start方法则调用
	input.Source = source
	if ss, ok := input.Source.(lifi.StreamingSource); ok {
		collector := lifi.NewCollector(dst)
		err := ss.Start(collector)
		if err != nil {
			return nil, fmt.Errorf("starting source %v", err)
		}
	}

	//如果有 Init 方法则调用
	if ss, ok := input.Source.(lifi.Initializer); ok {
		err := ss.Init()
		if err != nil {
			return nil, fmt.Errorf("starting source %v", err)
		}
	}

	unit.inputs = append(unit.inputs, input)

	return unit, nil
}

func (a *Pipe) startSink(
	ctx context.Context,
	runningSinks []models.RunningSink,
) (chan<- lifi.Data, *sinkUnit, error) {
	src := make(chan lifi.Data, 100)

	unit := &sinkUnit{
		src:   src,
		trans: make(map[string]transferUnits),
	}

	for _, sink := range runningSinks {
		creator, ok := sinks.Sinks[sink.Name]
		if !ok {
			return nil, nil, fmt.Errorf("undefined but requested sink: %s", sink.Name)
		}

		sk := creator()
		a.UnmarshalConfigJson(*sink.Config, sk)
		sink.Sink = sk
		//如果有 Init 方法则调用
		if ss, ok := sink.Sink.(lifi.Initializer); ok {
			err := ss.Init()
			if err != nil {
				return nil, nil, fmt.Errorf("initializing sink error: %v", err)
			}
		}
		// 设置序列化
		if ss, ok := sink.Sink.(lifi.SerializerPlugin); ok {
			ss.SetSerializer(&lifi.ByteDataSerializer{})
		}
		err := a.connectSink(ctx, sink)

		if err != nil {
			for _, s := range unit.sinks {
				err := s.Sink.Close()
				if err != nil {
				}
			}
			return nil, nil, fmt.Errorf("fiald connecting output  %w", err)

		}

		//启动transfer
		if len(sink.Transfers) > 0 {
			for _, t := range sink.Transfers {
				creator, ok := transfers.Transfers[t.Name]
				if !ok {
					return nil, nil, fmt.Errorf("undefined but requested transfer: %s", t.Name)
				}
				transfer := creator()

				if t.Config != nil {

					a.UnmarshalConfigJson(*t.Config, transfer)
				}
				//如果有 Init 方法则调用
				if tsf, ok := transfer.(lifi.Initializer); ok {
					err := tsf.Init()
					if err != nil {
						return nil, nil, fmt.Errorf("initializing sink error: %v", err)
					}
				}
				t.Transfer = transfer

			}

			// 按照transfer order 排序
			sort.Stable(sink.Transfers)

			// 按顺序启动transfer
			next, tu, err := a.startTransfer(src, sink.Transfers)
			if err != nil {
				return nil, nil, err
			}
			unit.trans[sink.Name] = tu

			src = next
		}

		unit.sinks = append(unit.sinks, sink)
	}

	return src, unit, nil
}

func (a *Pipe) startTransfer(dst chan lifi.Data, transfers []*models.RunningTransfer) (chan lifi.Data, []*transferUnit, error) {
	units := make([]*transferUnit, 0, len(transfers))
	var src chan lifi.Data

	//for i := 0; i < len(transfers); i++ {
	for i := len(transfers) - 1; i >= 0; i-- {
		transfer := transfers[i]
		src = make(chan lifi.Data, 100)
		acc := lifi.NewCollector(dst)

		err := transfer.Start(acc)
		if err != nil {
			for _, u := range units {
				u.transfer.Stop()
				close(u.dst)
			}
			return nil, nil, fmt.Errorf("starting processor  %v", err)
		}

		units = append(units, &transferUnit{
			src:      src,
			dst:      dst,
			transfer: transfer,
		})

		dst = src
	}

	return src, units, nil

}

func (a *Pipe) runSources(
	ctx context.Context,
	startTime time.Time,
	unit *sourceUnit) {
	var wg sync.WaitGroup
	log.Printf("Starting %d inputs", len(unit.inputs))
	log.Printf("start at %s", startTime.Format("2006-01-02 15:04:05"))
	for _, source := range unit.inputs {

		collect := lifi.NewCollector(unit.dst)
		interval := time.Duration(a.Pipeline.Interval) * time.Millisecond
		wg.Add(1)
		go func(input models.RunningSource) {
			defer wg.Done()
			a.gatherLoop(ctx, collect, input, interval)
		}(source)
	}

}
func (a *Pipe) gatherLoop(
	ctx context.Context,
	collect lifi.Collector,
	input models.RunningSource,
	interval time.Duration,
) {
	defer panicRecover(input)
	ticker := time.NewTicker(interval)
	for {
		select {
		case <-ticker.C:
			err := a.gatherOnce(input, collect, interval)
			if err != nil {
				log.Printf("err:%s", err)
			}
		case <-ctx.Done():
			return
		}
	}
}

func (a *Pipe) gatherOnce(input models.RunningSource, collect lifi.Collector, interval time.Duration) error {

	done := make(chan error)
	go func() {
		done <- input.Source.Gather(collect)
	}()

	slowWarning := time.NewTicker(interval)
	defer slowWarning.Stop()

	for {
		select {
		case err := <-done:
			return err
		case <-slowWarning.C:
			log.Printf("Collection took longer than expected; not complete after interval of %s",
				interval)
		}
	}

	return nil
}

func (a *Pipe) runTransfers(units []*transferUnit) {
	var wg sync.WaitGroup
	for _, unit := range units {
		wg.Add(1)
		go func(unit *transferUnit) {
			defer wg.Done()

			acc := lifi.NewCollector(unit.dst)
			for m := range unit.src {
				if err := unit.transfer.Process(m, acc); err != nil {
					acc.AddError(err)
					m.Drop()
				}
			}
			log.Printf("Transfer channel closing [%s]", unit.transfer.Name)
			unit.transfer.Stop()
			close(unit.dst)
			log.Printf("Transfer channel closed[%s]", unit.transfer.Name)
		}(unit)
	}
	wg.Wait()

}

func (a *Pipe) runSinks(unit *sinkUnit) {

	for _, output := range unit.sinks {
		tu := unit.trans[output.Name]
		go a.runTransfers(tu)
	}

	for data := range unit.src {
		for _, output := range unit.sinks {
			// todo output.batch

			err := output.Sink.Write(data)
			if err != nil {
				log.Printf("sink[%s] write error:%v", output.Name, err)
			}

		}
	}

}

func (a *Pipe) UnmarshalConfig(cfg lifi.PluginConfig, v interface{}) {

	yamlData, err := yaml.Marshal(cfg)
	if err != nil {
		log.Println("yaml marshal error:", err)
		panic(err)
	}

	// 将 yaml 字符串解析到结构体
	err = yaml.Unmarshal(yamlData, v)
	if err != nil {
		log.Println("yaml unmarshal error:", err)
		panic(err)

	}
}

func (a *Pipe) UnmarshalConfigJson(cfg lifi.PluginConfig, v interface{}) {

	jsonData, err := json.Marshal(cfg)
	if err != nil {
		log.Println("json marshal error:", err)
		panic(err)
	}

	// 将 yaml 字符串解析到结构体
	err = json.Unmarshal(jsonData, v)
	if err != nil {
		log.Println("json unmarshal error:", err)
		panic(err)

	}
}

// panicRecover displays an error if an input panics.
func panicRecover(input models.RunningSource) {
	if err := recover(); err != nil {
		trace := make([]byte, 2048)
		runtime.Stack(trace, true)
		log.Printf("input %s panicked: %s\n%s", input.Name, err, trace)
	}
}
